package com.hp.bon.sgw.dr;

import java.util.HashSet;
import java.util.Iterator;
import java.util.LinkedList;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;

import com.hp.bon.sgw.bean.SGWStatQueueBean;
import com.hp.bon.sgw.bean.SysParamBean;
import com.hp.bon.sgw.service.ConfigListener;
import com.hp.bon.sgw.service.ConfigService;
import com.hp.bon.sgw.start.SpringContext;
import com.hp.bon.sgw.util.Comutils;
import com.hp.mdr.api.proxy.MdrProxyApi;
import com.hp.mdr.api.proxy.MdrProxySendInfo;

/**
 * drproxy 将DCC消息保存在这个queue中，可能是本地drproxy收到的消息CCR，也可能远端drproxy发来的消息
 * 这些消息正在处理中，还未完成处理。
 * 
 * @author Sandy
 * 
 */
public class DrproxyMessageQueue implements ConfigListener {
	/**
	 * Logger for this class
	 */
	private static final Logger logger = LoggerFactory.getLogger(DrproxyMessageQueue.class);
	/**
	 * 当前允许的消息处理最大并发数
	 */
	private volatile int maxInprogressCount = 10000;
	/**
	 * DCCProgress 超时时间（秒）
	 */
	private volatile int dccProgressTimeoutSeconds = 300;
	
	private volatile int drproxyTimeoutSeconds = 259200; //72小时

	/**
	 * 守护线程扫描的间隔时间
	 */
	private volatile int watchDogScanSecondes = 1;
	/**
	 * 队列容量告警阀值
	 */
	private volatile int queueWarnThreshold = 66;

	private ConfigService confServ;

	/**
	 * 保存当前正在处理中的消息状态, Key=配对串
	 */
	public static ConcurrentHashMap<String, DrCallParam> msgProgressMap = new ConcurrentHashMap<String, DrCallParam>();
	
	/**
	 * key = session ID，只有一种情况会用到这个map，当消息来自主业务系统，drpoxy必须删除auditKey，才能发送CCR到外部系统，外部系统返回的时候，request中无法获取auditkey
	 * 此情况下用sessionId作为主键
	 */
	public static ConcurrentHashMap<String, DrCallParam> msgSessionProgressMap = new ConcurrentHashMap<String, DrCallParam>();

	WatchDogThread watchDogThread;
	WatchDogThread watchDogSessionThread;

	/**
	 * 所关心的运行期变化的配置数据的Key
	 */
	private final Set<String> monitorConfKeys = new HashSet<String>();
	private volatile long prevQuequeWarnedTime;

	/**
	 * 配置同DrproxyMessageQueue的配置
	 */
	public DrproxyMessageQueue() {
		monitorConfKeys.add(SysParamBean.KEY_AGREEMENT_Timeout_LinkWait);
		monitorConfKeys.add("InprogressMessageQueue.Max");
		monitorConfKeys.add("InprogressMessageQueue.WarnThreshold");
		monitorConfKeys.add("InprogressMessageQueue.WatchDogScanSconds");
	}

	static DrproxyMessageQueue instance;

	public static DrproxyMessageQueue getInstance() {
		if (instance == null) {
			instance = SpringContext.getApplicationContext().getBean(DrproxyMessageQueue.class);
		}
		return instance;
	}

	/**
	 * 初始化过程
	 */
	public void init() {
		loadConfParams();
		watchDogThread = new WatchDogThread();
		watchDogThread.start();
		
	}

	@Override
	public void onConfigUpdated(String updatedKey) {
		loadConfParams();
	}

	private void loadConfParams() {
		String paramName = SysParamBean.KEY_AGREEMENT_Timeout_LinkWait;
		dccProgressTimeoutSeconds = confServ.getIntValue(paramName, 60);
		logger.info("conf param " + paramName + " value " + dccProgressTimeoutSeconds);
		paramName = "InprogressMessageQueue.Max";
		maxInprogressCount = confServ.getIntValue(paramName, 10000);
		logger.info("conf param " + paramName + " value " + maxInprogressCount);
		paramName = "InprogressMessageQueue.WarnThreshold";
		queueWarnThreshold = confServ.getIntValue(paramName, 80);
		logger.info("conf param " + paramName + " value " + queueWarnThreshold);
		paramName = "InprogressMessageQueue.WatchDogScanSeconds";
		watchDogScanSecondes = confServ.getIntValue(paramName, 1);
		logger.info("conf param " + paramName + " value " + watchDogScanSecondes);

	}

	/**
	 * 队列使用率 0-100之间
	 * 
	 * @return
	 */
	public double getQueueUsage() {
		return 100 * (msgProgressMap.size() + 0.0) / maxInprogressCount;
	}

	/**
	 * 取队列最大长度
	 * 
	 * @return
	 */
	public int getMaxInprogressCount() {
		return maxInprogressCount;
	}

	/**
	 * 第一个元素是队列当前长度，第二个元素是所有请求的数据字节长度总和
	 * 
	 * @return
	 */
	public int getQueueSize() {
		return msgProgressMap.values().size();
	}

	/**
	 * 添加一个正在处理的新请求
	 * 
	 * @param msgProgress
	 */
	public void addRequest(DrCallParam drCallParam) {
		if (drCallParam.getTimeOutTime() == 0) {
			// 设置超时时间 72小时
			int timeout = drproxyTimeoutSeconds;

			drCallParam.setTimeOutTime(System.currentTimeMillis() + timeout * 1000);
			//logger.info(callParam.getSessionId() + " timeout:" + Comutils.formatTime(callParam.getTimeOutTime(), "yyyyMMdd hh:mm:ss.SSS") + ":" + timeout + "s");
		}
		msgProgressMap.put(drCallParam.getAuditKey(),drCallParam);
	}
	
	public void addSessionRequest(DrCallParam drCallParam) {
		if (drCallParam.getTimeOutTime() == 0) {
			// 如果没有设置超时时间
			int timeout = drproxyTimeoutSeconds;

			drCallParam.setTimeOutTime(System.currentTimeMillis() + timeout * 1000);
			//logger.info(callParam.getSessionId() + " timeout:" + Comutils.formatTime(callParam.getTimeOutTime(), "yyyyMMdd hh:mm:ss.SSS") + ":" + timeout + "s");
		}
		msgSessionProgressMap.put(drCallParam.getSessionId(),drCallParam);
	}

	/**
	 * 将原有的某个msg process移除
	 * 
	 * @param oldMsgID
	 */
	public DrCallParam removeProgress(String oldMsgID) {
		return msgProgressMap.remove(oldMsgID);
	}
	
	public DrCallParam removeSessionProgress(String oldMsgID) {
		return msgSessionProgressMap.remove(oldMsgID);
	}

	@Autowired(required = true)
	public void setConfServ(ConfigService confServ) {
		this.confServ = confServ;
		confServ.addConfigListener(this);
	}

	public SGWStatQueueBean getQueueStatusBean() {
		SGWStatQueueBean bean = new SGWStatQueueBean();
		bean.setMaxQueue(this.maxInprogressCount);
		bean.setUsedQueue(msgProgressMap.size());
		bean.setThreshold(this.queueWarnThreshold);
		long warnedTime = this.prevQuequeWarnedTime;
		if (warnedTime != 0) {
			bean.setPrevWarnedTime(Comutils.formatTime(warnedTime, Comutils.GEN_DATA_FORMAT));
		} else {
			bean.setPrevWarnedTime("");
		}
		return bean;
	}

	public int getPendingTotals() {
		return msgProgressMap.size();
	}

	/**
	 * 得到当前队列中各种请求的Pending的消息数目
	 * 
	 * @return 队列长度
	 */
	public int[] getPendingCounts() {
		int[] totalCounts = new int[3];
		
		return totalCounts;
	}

	/**
	 * 得到当前队列中的能力数量统计信息
	 * 
	 * @return
	 */
	public int getCurCapsSum(Map<String, Integer> sumMap) {
		int totalCount = 0;
		for (DrCallParam progrss : msgProgressMap.values()) {
			totalCount++;
		}
		return totalCount;

	}

	public void shutdown() {
		watchDogThread.interrupt();
	}

  public Set<String> monitorConfKeys()
  {
    return this.monitorConfKeys;
  }

  public String getQueneContext() {
    StringBuilder sb = new StringBuilder();
    Iterator itor = msgProgressMap.entrySet().iterator();
    Map.Entry entry = null;

    while (itor.hasNext()) {
      entry = (Map.Entry)itor.next();
      sb.append((String)entry.getKey() + " = " + Comutils.formatTime(((DrCallParam)entry.getValue()).getTimeOutTime(), "yyyy-MM-dd hh:mm:ss.SSS"));
      sb.append("\r\n");
    }

    return sb.toString();
  }

  class WatchDogSessionThread extends Thread
  {
    public WatchDogSessionThread()
    {
      setName("DrproxyMessageQueue-WatchDogSessionThread");
      setDaemon(true);
    }

    public void run() {
      while (!Thread.interrupted()) {
        checkSessionQueue();
        try {
          Thread.sleep(DrproxyMessageQueue.this.watchDogScanSecondes * 1000);
        }
        catch (Exception localException) {
        }
      }
      DrproxyMessageQueue.logger.info("shutdowned");
    }

    private void checkSessionQueue()
    {
      try
      {
        int queueLen = 0;
        Iterator itor = DrproxyMessageQueue.msgSessionProgressMap.entrySet().iterator();
        long curTime = System.currentTimeMillis();
        while (itor.hasNext()) {
          queueLen++;
          Map.Entry entry = (Map.Entry)itor.next();
          DrCallParam msgPrgrss = (DrCallParam)entry.getValue();
          if (msgPrgrss.getTimeOutTime() < curTime) {
            DrproxyMessageQueue.logger.warn("17004|found message process timeout,auditKey=" + msgPrgrss.getAuditKey());
            msgPrgrss.setRetCode(11203);
            msgPrgrss.setResultMsg("BON TimeOut");

            itor.remove();
            DrproxyMessageQueue.logger.info("CCR from master business NE timeout. CCR is removed from msgSessionProgressMap.");
          }
        }

        if ((queueLen * DrproxyMessageQueue.this.queueWarnThreshold / 100 > DrproxyMessageQueue.this.maxInprogressCount) && (DrproxyMessageQueue.this.prevQuequeWarnedTime + 300000L < curTime)) {
          DrproxyMessageQueue.logger.warn("17005|current queue reached threshold " + DrproxyMessageQueue.this.queueWarnThreshold + ",cur length:" + queueLen + ",max capacity:" + DrproxyMessageQueue.this.maxInprogressCount);
          DrproxyMessageQueue.this.prevQuequeWarnedTime = curTime;
        }
      } catch (Exception e) {
        DrproxyMessageQueue.logger.warn("checkQueue:" + e.toString());
      }
    }
  }

  class WatchDogThread extends Thread
  {
    public WatchDogThread()
    {
      setName("DrproxyMessageQueue-WatchDogThread");
      setDaemon(true);
    }

    public void run() {
      while (!Thread.interrupted())
      {
        checkQueue();
        try {
          Thread.sleep(DrproxyMessageQueue.this.watchDogScanSecondes * 1000);
        }
        catch (Exception localException) {
        }
      }
      DrproxyMessageQueue.logger.info("shutdowned");
    }

    private void checkQueue()
    {
      try
      {
        int queueLen = 0;
        Iterator itor = DrproxyMessageQueue.msgProgressMap.entrySet().iterator();
        long curTime = System.currentTimeMillis();
        List<MdrProxySendInfo> send_info_list_ = new LinkedList();
        while (itor.hasNext()) {
          queueLen++;
          Map.Entry entry = (Map.Entry)itor.next();
          DrCallParam msgPrgrss = (DrCallParam)entry.getValue();

          if (msgPrgrss.getTimeOutTime() < curTime) {
            DrproxyMessageQueue.logger.warn("17004|found message process timeout,auditKey=" + msgPrgrss.getAuditKey());
            msgPrgrss.setRetCode(11203);
            msgPrgrss.setResultMsg("BON TimeOut");

            itor.remove();

            DrproxyMessageQueue.logger.info("CCR in drproxyMessageQueue is timeout. CCR is removed from msgProgressMap.");
            msgPrgrss.getDrProcessor().endProcess(msgPrgrss);
          }
          else if (msgPrgrss.isWait4Resend())
          {
            msgPrgrss.getDrProcessor().resend2Slave(msgPrgrss);
            if ((!msgPrgrss.isWait4Resend()) && (msgPrgrss.getToPeerHostId().equals(DrproxyUtils.otherDrproxyHostId()))) {
              itor.remove();
            }

          }
          else
          {
            send_info_list_.clear();
            String toHostId = msgPrgrss.getToPeerHostId();
            int rc_ = MdrProxyApi.mdr_QuerySend(toHostId, send_info_list_);
            if (rc_ != 0) {
              DrproxyMessageQueue.logger.error("call MdrProxyApi.mdr_QuerySend error.");
            }
            else {
              for (MdrProxySendInfo elem_ : send_info_list_) {
                DrCallParam drCallParam = 
                  (DrCallParam)DrproxyMessageQueue.msgProgressMap
                  .get(elem_.auditKey);
                if (drCallParam != null) {
                  drCallParam.getDrProcessor().process(
                    drCallParam);
                }
              }

            }

          }

        }

        if ((queueLen * DrproxyMessageQueue.this.queueWarnThreshold / 100 > DrproxyMessageQueue.this.maxInprogressCount) && (DrproxyMessageQueue.this.prevQuequeWarnedTime + 300000L < curTime)) {
          DrproxyMessageQueue.logger.warn("17005|current queue reached threshold " + DrproxyMessageQueue.this.queueWarnThreshold + ",cur length:" + queueLen + ",max capacity:" + DrproxyMessageQueue.this.maxInprogressCount);
          DrproxyMessageQueue.this.prevQuequeWarnedTime = curTime;
        }
      } catch (Exception e) {
        DrproxyMessageQueue.logger.warn("checkQueue:" + e.toString());
      }
    }
  }
}